
#include <iostream>
#include <fstream>
#include <map>
#include <vector>
#include <string>

#include <client_environment.hpp>
#include <io_service_pool_impl.hpp>

#include <xbase_utility.hpp>
#include <dynamic_buffer.hpp>

struct insert_item
{
	std::string m_cid;
	std::string m_peerid;
	std::string m_gcid;
	uint64_t m_filesize;
	uint64_t m_timestamp;

	std::istream &load(std::istream &in) {
		in >> m_cid >> m_peerid
		   >> m_gcid >> m_filesize
		   >> m_timestamp;
		return in;
	}
};

typedef std::vector<insert_item> item_vector;

// class row_mutator_pool
// 	: private boost::noncopyable
// {
// public:
// 	row_mutator_pool(xbase::client_environment &env,
// 			 std::size_t max_size = 10000)
// 		: m_env(env)
// 		, m_max_size(max_size) {}

// 	xbase::row_mutator_ptr new_row_mutator() {
// 		return (m_free_mutator.empty()
// 			? m_env.new_row_mutator()
// 			: m_free_mutator.pop_back());
// 	}

// 	void delete_row_mutator(xbase::row_mutator_ptr ptr) {
// 		if(m_free_mutator.size() < m_max_size)
// 			m_free_mutator.push_front(ptr);
// 	}

// private:
// 	xbase::client_environment &m_env;
// 	std::deque<xbase::row_mutator_ptr> m_free_mutator;
// 	std::size_t m_max_size;
// };

// class row_mutation_pool
// 	: private boost::noncopyable
// {
// public:
// 	row_mutator_pool(xbase::protocol::table_info_ptr table,
// 			 std::size_t max_size = 10000)
// 		: m_table_info(table)
// 		, m_max_size(max_size) {}

// 	xbase::row_mutation new_row_mutation(const std::string &row_name) {
// 		if(m_free_mutation.empty())
// 		{
// 			return xbase::row_mutation(m_table_info, row_name);
// 		}
// 		else
// 		{
// 			xbase::row_mutation_data_ptr ptr = m_free_mutation.pop_back();
// 			ptr->change_row_name(row_name);
// 			return xbase::row_mutation(m_table_info, ptr);
// 		}
// 	}

// 	void delete_row_mutation(xbase::row_mutation mutation) {
// 		if(m_free_mutation.size() < m_max_size)
// 		{
// 			mutation.clear();
// 			m_free_mutation.push_front(mutation.data_ptr());
// 		}
// 	}

// private:
// 	xbase::protocol::table_info_ptr m_table_info;
// 	std::deque<xbase::row_mutation_data_ptr> m_free_mutation;
// 	std::size_t m_max_size;
// };

std::size_t count_commit = 0;
std::size_t count_complete = 0;
std::size_t count_failed = 0;

void on_complete(bool verbos, std::size_t show_count,
		 const boost::system::error_code &ec)
{
	if(ec)
	{
		++ count_failed;
		if(verbos)
		{
			std::cout << "apply data failed: " << ec
				  << " - " << ec.message() << std::endl;
		}
	}
	++count_complete;
	if(verbos && (count_complete % show_count == 0))
	{
		std::cout << "count_complete = " << count_complete
			  << ", count_failed = " << count_failed
			  << std::endl;
	}
}

void on_complete_multi_row(xbase::multi_row_mutator::container_ptr rows,
			   bool verbos, std::size_t show_count,
			   std::size_t success_row, const boost::system::error_code &ec)
{
	if(ec)
	{
		count_failed += rows->size() - success_row;
		if(verbos)
		{
			std::cout << "apply data failed: success_row = " << success_row
				  << ", ec = " << ec
				  << " - " << ec.message() << std::endl;
		}
	}
	count_complete += rows->size();
	if(verbos && (count_complete % show_count == 0))
	{
		std::cout << "count_complete = " << count_complete
			  << ", count_failed = " << count_failed
			  << std::endl;
	}
}

void insert_data(xbase::client_environment &env,
		 const xbase::protocol::table_info_ptr &table,
		 const std::string &column_name,
		 const item_vector &vec,
		 bool verbos, std::size_t show_count,
		 bool multi_row_mode, std::size_t max_row_num_in_multi_row_mode)
{
	xbase::multi_row_mutator::container_ptr rows(new xbase::multi_row_mutator::container_type);
	
	for(item_vector::const_iterator it = vec.begin();
	    it != vec.end(); ++ it)
	{
			std::string row_data = dump_string_unhex(it->m_cid);
			row_data += it->m_peerid;

			std::string value_data = dump_string_unhex(it->m_gcid);
			value_data.append(reinterpret_cast<const char*>(&it->m_filesize),
					  sizeof(it->m_filesize));

			xbase::row_mutation mutation(table, row_data);
			mutation.add(column_name, value_data);

		if(! multi_row_mode)
		{
			env.async_apply(mutation,
					boost::bind(on_complete,
						    verbos, show_count, _1));

			++count_commit;
			if(verbos && (count_commit % show_count == 0))
			{
				std::cout << "count_commit = " << count_commit
					  << std::endl;
			}
		}
		else
		{
			rows->push_back(mutation.data_ptr());
			if(rows->size() >= max_row_num_in_multi_row_mode)
			{
				std::sort(rows->begin(), rows->end(), &xbase::protocol::row_mutation_data_ptr_less);
				env.async_apply(table->m_table_name,
						rows,
						boost::bind(on_complete_multi_row, rows,
							    verbos, show_count, _1, _2));
				count_commit += rows->size();
				rows.reset(new xbase::multi_row_mutator::container_type);
				if(verbos && (count_commit % show_count == 0))
				{
					std::cout << "count_commit = " << count_commit
						  << std::endl;
				}
			}
		}
	}
}

int main(int argc, char **argv)
{
	LOG_CONFIG("../../conf/log4cplus.tablet_server.conf");

	if((argc < 4) || (argc > 12))
	{
		std::cout << "test_insert_imphub_data <master ip> <master port> <data file> "
			  << " [table_name=inmedia]"
			  << " [column_name='gcid:_'] "
			  << " [cache_count=100000]"
			  << " [verbos/nonverbos]"
			  << " [show_count=10000]"
			  << " [speed_limit=0]"
			  << " [single_row_mode/multi_row_mode]"
			  << " [multi_row_num=1000]"
			  << std::endl;
		return 1;
	}

	const boost::asio::ip::tcp::endpoint master_ep(
		boost::asio::ip::address::from_string(argv[1]),
		::atoi(argv[2])
		);
	std::ifstream infile(argv[3]);
	if(! infile)
	{
		std::cout << "open file " << argv[3] << " failed" << std::endl;
		return 1;
	}
	const char *table_name = (argc >= 5 ? argv[4] : "inmedia");
	const std::string column_name = (argc >= 6 ? argv[5] : "gcid:_");

	std::size_t counter_item = 0;
	uint64_t counter_time = 0;
	uint64_t adjust_time = 0;

	xbase::async_module_stats::enable();
	xbase::client_options_ptr options(new xbase::client_options);
	xbase::io_service_pool_ptr pool(new xbase::impl::single_io_service);
	xbase::client_environment env(options, pool, master_ep);

	xbase::table_info_cache::sync_handler_type sh;
	env.async_open_table(table_name, sh.make_handler());
	sh.wait();
	if(sh.get_error_code())
	{
		std::cerr << "open table failed: " << sh.get_error_code().message() << std::endl;
		return 0;
	}
	xbase::protocol::table_info_ptr table_info = sh.get_table_info();

	std::size_t cache_count = 100000;
	bool verbos = true;
	std::size_t show_count = 10000;
	std::size_t speed_limit = 0; // 每秒限制多少条写命令
	if(argc >= 7)
	{
		cache_count = ::atoi(argv[6]);
	}
	if(argc >= 8)
	{
		verbos = (std::string("verbos") == argv[7]);
	}
	if(argc >= 9)
	{
		show_count = ::atoi(argv[8]);
	}
	if(argc >= 10)
	{
		speed_limit = ::atoi(argv[9]);
	}

	bool multi_row_mode = false;
	if(argc >= 11)
	{
		multi_row_mode = (std::string("multi_row_mode") == argv[10]);
	}
	std::size_t multi_row_num = 1000;
	if(argc >= 12)
	{
		multi_row_num = ::atoi(argv[11]);
	}

	insert_item item;
	item_vector item_vec;
	item_vec.reserve(cache_count + 10000);
	while((! infile.eof()) || (! item_vec.empty()))
	{
		if(! infile.eof())
		{
			item.load(infile);

			if(! infile.eof())
			{
				item_vec.push_back(item);

				++ counter_item;
			}
		}

		if((item_vec.size() >= cache_count) ||
		   (infile.eof() && (! item_vec.empty())))
		{
			if(verbos)
				std::cout << "applying ..." << std::endl;
			
			const uint64_t t_start = cur_micro_seconds();

			insert_data(env, table_info, column_name, item_vec,
				    verbos, show_count,
				    multi_row_mode, multi_row_num);
			
			const uint64_t t_end = cur_micro_seconds();
			const uint64_t t_used = t_end - t_start;
			counter_time += t_used;

			if(verbos)
			{
				const double used = t_used / (1000.0 * 1000.0);
				std::cout << item_vec.size() << " items applied, used "
					  << used << " seconds, totally "
					  << counter_item << " items applied, speed at "
					  << ((double)counter_item) * 1000.0 * 1000.0 / ((double)counter_time)
					  << " items per second, adjusted speed is "
					  << ((double)counter_item) * 1000.0 * 1000.0 / ((double)(counter_time + adjust_time))
					  << std::endl;
				std::cout << "commit info: count_commit = " << count_commit
					  << ", count_complete = " << count_complete
					  << ", count_failed = " << count_failed << std::endl;
			}

			item_vec.clear();

			if(speed_limit != 0)
			{
				uint64_t counter = static_cast<uint64_t>(counter_item) * 1000 * 1000;
				uint64_t expect_time = counter / speed_limit;
				if((counter_time + adjust_time) < expect_time)
				{
					const uint64_t sleep_time = expect_time - (counter_time + adjust_time);
					adjust_time += sleep_time;
					if(verbos)
					{
						std::cout << "adjust speed by sleep " << sleep_time << " micro seconds" << std::endl;
					}
					usleep(static_cast<useconds_t>(sleep_time));
				}
			}
		}
	}

	std::cout << "--- --- ---" << std::endl << "totally "
		  << counter_item << " items applied, used "
		  << (double)counter_time / (1000.0 * 1000.0) << " seconds, speed at "
		  << ((double)counter_item) * 1000.0 * 1000.0 / ((double)counter_time)
		  << " items per second"
		  << std::endl;

	if(count_complete < count_commit)
	{
		std::cout << "waiting complete ... " << std::endl;;
		while(count_complete < count_commit)
		{
			std::cout << "commit info: count_commit = " << count_commit
				  << ", count_complete = " << count_complete
				  << ", count_failed = " << count_failed << std::endl;
			sleep(3);
		}
		std::cout << "commit info: count_commit = " << count_commit
			  << ", count_complete = " << count_complete
			  << ", count_failed = " << count_failed << std::endl;
	}

	env.stop();
	env.join();
	xbase::async_module_stats::print(std::cout);

	return 0;
}

